package com.atguigu.edu.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author 杨晨昱
 */
public class DwdTrafficBaseLogSplit {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "topic_log";
        String groupId = "dwd_traffic_base_log_split_group";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaDSStream = env.addSource(kafkaConsumer);

        OutputTag<String> dirtyTag = new OutputTag<String>("dirtyTag"){};
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDSStream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObj = JSON.parseObject(value);
                    out.collect(jsonObj);
                } catch (Exception e) {
                    ctx.output(dirtyTag, value);
                }
            }
        });
        DataStream<String> dirtyDS = jsonObjDS.getSideOutput(dirtyTag);
        dirtyDS.addSink(MyKafkaUtil.getKafkaProducer("dirty_data"));

        KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"));
        SingleOutputStreamOperator<JSONObject> fixedDS = keyedDS.map(new RichMapFunction<JSONObject, JSONObject>() {
            private ValueState<String> lastViewTime;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("lastViewTime", String.class);
                valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite).build());
                this.lastViewTime = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public JSONObject map(JSONObject jsonObj) throws Exception {
                String isNew = jsonObj.getJSONObject("common").getString("is_new");
                String lastView = lastViewTime.value();
                Long ts = jsonObj.getLong("ts");
                String date = DateFormatUtil.toDate(ts);
                if ("1".equals(isNew)) {
                    if (StringUtils.isEmpty(lastView)) {
                        lastViewTime.update(date);
                    } else {
                        if (!lastView.equals(date)) {
                            isNew = "0";
                            jsonObj.getJSONObject("common").put("is_new", isNew);
                        }
                    }


                } else {
                    String s = DateFormatUtil.toDate(ts - 24 * 3600 * 1000);
                    lastViewTime.update(s);

                }


                return jsonObj;
            }
        });
        fixedDS.print();


        OutputTag<String> errTag = new OutputTag<String>("errTag"){};
        OutputTag<String> startTag = new OutputTag<String>("startTag"){};
        OutputTag<String> displayTag = new OutputTag<String>("displays") {};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag"){};
        OutputTag<String> appVideoTag = new OutputTag<String>("appVideo"){};


        SingleOutputStreamOperator<String> pageDS = fixedDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObj, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                //错误日志
                JSONObject errJsonObj = jsonObj.getJSONObject("err");
                if (errJsonObj != null) {
                    ctx.output(errTag, jsonObj.toJSONString());
                    jsonObj.remove("err");
                }
                JSONObject startJsonObj = jsonObj.getJSONObject("start");
                if (startJsonObj != null) {
                    ctx.output(startTag, jsonObj.toJSONString());
                } else if (jsonObj.getJSONObject("appVideo") != null) {
                    ctx.output(appVideoTag, jsonObj.toJSONString());
                } else {
                    JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                    JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                    Long ts = jsonObj.getLong("ts");

                    JSONArray displaysArr = jsonObj.getJSONArray("displays");
                    if (displaysArr != null && displaysArr.size() > 0) {
                        for (int i = 0; i < displaysArr.size(); i++) {
                            JSONObject newDisplayJsonObj = displaysArr.getJSONObject(i);
                            JSONObject displayobj = new JSONObject();
                            displayobj.put("common", commonJsonObj);
                            displayobj.put("ts", ts);
                            displayobj.put("display", newDisplayJsonObj);
                            displayobj.put("page", pageJsonObj);
                            ctx.output(displayTag, displayobj.toJSONString());
                        }
                        jsonObj.remove("displays");
                    }

                    JSONArray actionsJson = jsonObj.getJSONArray("actions");
                    if (actionsJson != null && actionsJson.size() > 0) {
                        for (int i = 0; i < actionsJson.size(); i++) {
                            JSONObject newActionJsonObj = new JSONObject();
                            JSONObject actionsJsonJSONObj = actionsJson.getJSONObject(i);
                            newActionJsonObj.put("ts", ts);
                            newActionJsonObj.put("action", actionsJsonJSONObj);
                            newActionJsonObj.put("page", pageJsonObj);
                            newActionJsonObj.put("common", commonJsonObj);
                            ctx.output(actionTag, newActionJsonObj.toJSONString());

                        }
                        jsonObj.remove("actions");
                    }
                    out.collect(jsonObj.toJSONString());
                }
            }
        });
        DataStream<String> errDS = pageDS.getSideOutput(errTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);
        DataStream<String> actionDS = pageDS.getSideOutput(actionTag);
        DataStream<String> appVideoDS = pageDS.getSideOutput(appVideoTag);
        DataStream<String> startDS = pageDS.getSideOutput(startTag);

        pageDS.print(">>>>>");
        errDS.print("###");
        startDS.print("@@@");
        displayDS.print("$$$");
        actionDS.print("^^^^");

        pageDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_page_log"));
        errDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_err_log"));
        startDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_start_log"));
        displayDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_display_log"));
        actionDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_action_log"));
        appVideoDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_appVideo_log"));
        env.execute();
    }

}
